package org.xq.softcup.mq.admin.service.impl;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.springframework.web.context.request.async.DeferredResult;
import org.xq.softcup.mq.admin.model.RegData;
import org.xq.softcup.mq.admin.model.Registry;
import org.xq.softcup.mq.admin.util.FastJsonUtil;
import org.xq.softcup.mq.admin.util.JsonResult;
import org.xq.softcup.mq.admin.util.JacksonUtil;
import org.xq.softcup.mq.admin.util.PropUtil;
import org.xq.softcup.mq.admin.dao.IRegDao;
import org.xq.softcup.mq.admin.dao.IRegDataDao;
import org.xq.softcup.mq.admin.service.IMqRegService;

import javax.annotation.Resource;
import java.io.File;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 注册服务实现类
 *
 * @author admin
 */
@Slf4j
@Service
public class MqRegServiceImpl implements IMqRegService, InitializingBean, DisposableBean {

    @Resource
    private IRegDao regDao;
    @Resource
    private IRegDataDao regDataDao;
//    @Resource
//    private IRegMsgDao regMsgDao;
    //目前当前路径
    private String rootPath = System.getProperty("user.dir");
    //注册数据保存的文件路径
    @Value("${mq.registry.data.filepath}")
    private String regDataFilePath ;
    //心跳间隔
    @Value("${mq.registry.beattime}")
    private int beatTime;
    //访问令牌
    @Value("${mq.registry.accessToken}")
    private String accessToken;

    //广播和文件数据
    private ExecutorService executorService = Executors.newCachedThreadPool();
    private volatile boolean executorStoped = false;
    private volatile List<Integer> readedMessageIds = Collections.synchronizedList(new ArrayList<Integer>());

    private volatile LinkedBlockingQueue<RegData> registryQueue = new LinkedBlockingQueue<RegData>();
    private volatile LinkedBlockingQueue<RegData> removeQueue = new LinkedBlockingQueue<RegData>();
    private Map<String, List<DeferredResult>> registryDeferredResultMap = new ConcurrentHashMap<>();
    private Lock registryLock = new ReentrantLock();
    public static RegData staticRegData;

    /**
     * 服务注册方法, 添加的队列中, 在子线程中执行
     *
     * @param accessToken
     * @param regDataList
     * @return
     */
    @Override
    public JsonResult<String> registry(String accessToken, List<RegData> regDataList) {
        if (!StringUtils.isEmpty(this.accessToken) && !this.accessToken.equals(accessToken)) {
            return new JsonResult<String>(JsonResult.FAIL_CODE, "AccessToken 无效");
        }
        if (regDataList == null || regDataList.size() == 0) {
            return new JsonResult<String>(JsonResult.FAIL_CODE, "RegData 无效");
        }

        //添加到注册队列
        registryQueue.addAll(regDataList);
        return JsonResult.SUCCESS;
    }

    /**
     * 删除服务方法, 添加的队列中, 在子线程中执行
     *
     * @param accessToken
     * @param regDataList
     * @return
     */
    @Override
    public JsonResult<String> remove(String accessToken, List<RegData> regDataList) {
        if (!StringUtils.isEmpty(this.accessToken) && !this.accessToken.equals(accessToken)) {
            return new JsonResult<String>(JsonResult.FAIL_CODE, "AccessToken 无效");
        }
        if (regDataList == null || regDataList.size() == 0) {
            return new JsonResult<String>(JsonResult.FAIL_CODE, "RegData 无效");
        }

        // 添加删除队列
        removeQueue.addAll(regDataList);
        return JsonResult.SUCCESS;
    }

    @Override
    public JsonResult<Map<String, List<String>>> discovery(String accessToken, List<String> keys) {
        // valid
        if (!StringUtils.isEmpty(this.accessToken) && !this.accessToken.equals(accessToken)) {
            return new JsonResult<>(JsonResult.FAIL_CODE, "AccessToken 无效");
        }
        if (keys == null || keys.size() == 0) {
            return new JsonResult<>(JsonResult.FAIL_CODE, "keys 无效");
        }

        Map<String, List<String>> result = new HashMap<String, List<String>>();
        for (String key : keys) {
            RegData regData = new RegData();
            regData.setKey(key);
            List<String> dataList = new ArrayList<String>();
            Registry fileRegistry = getFileRegistryData(regData);
            if (fileRegistry != null) {
                dataList = fileRegistry.getDataList();
            }

            result.put(key, dataList);
        }
        return new JsonResult<Map<String, List<String>>>(result);
    }

    @Override
    public DeferredResult<JsonResult<String>> monitor(String accessToken, List<String> keys) {
        DeferredResult deferredResult = new DeferredResult(beatTime * 3 * 1000L, new JsonResult<>(JsonResult.SUCCESS_CODE, "Monitor timeout."));
        if (!StringUtils.isEmpty(this.accessToken) && !this.accessToken.equals(accessToken)) {
            deferredResult.setResult(new JsonResult<>(JsonResult.FAIL_CODE, "AccessToken 无效"));
            return deferredResult;
        }
        if (keys == null || keys.size() == 0) {
            deferredResult.setResult(new JsonResult<>(JsonResult.FAIL_CODE, "keys 无效"));
            return deferredResult;
        }
        for (String key : keys) {
            String fileName = parseRegistryDataFileName(key);

            List<DeferredResult> deferredResultList = registryDeferredResultMap.get(fileName);
            if (deferredResultList == null) {
                deferredResultList = new ArrayList<>();
                registryDeferredResultMap.put(fileName, deferredResultList);
            }
            deferredResultList.add(deferredResult);
        }

        return deferredResult;
    }

    /**
     * 更新注册信息并发送消息
     */
    private void checkRegistryData(RegData regData) {
        List<RegData> regDataList = regDataDao.findData(regData.getKey());
        List<String> valueList = new ArrayList<>();
        if (regDataList != null && regDataList.size() > 0) {
            for (RegData dataItem : regDataList) {
                valueList.add(dataItem.getValue());
            }
        }
        String dataJson = FastJsonUtil.objectToJSON(valueList);

        // 更新注册和消息
        Registry registry = regDao.load(regData.getKey());
        if (registry == null) {
            registry = new Registry();
            registry.setKey(regData.getKey());
            registry.setData(dataJson);
            regDao.add(registry);
        } else {
            if (!registry.getData().equals(dataJson)) {
                registry.setData(dataJson);
                regDao.update(registry);
            }
        }

    }

    /**
     * 发送注册数据和消息
    private void sendRegistryDataUpdateMessage(Registry rpcRegistry) {
        String registryUpdateJson = FastJsonUtil.objectToJSON(rpcRegistry);

        RegMsg regMsg = new RegMsg();
        regMsg.setData(registryUpdateJson);
        regMsgDao.add(regMsg);
    }
     */

    @Override
    public void afterPropertiesSet() throws Exception {
        /**
         * 使用异步线程注册
         */
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                while (!executorStoped) {
                    try {
                        RegData regData = registryQueue.take();
                        if (regData != null) {
                            //刷新和添加
                            int ret = regDataDao.refresh(regData);
                            if (ret == 0) {
                                regDataDao.add(regData);
                            }
                            //检查注册文件
                            Registry fileRegistry = getFileRegistryData(regData);
                            if (fileRegistry != null && fileRegistry.getDataList().contains(regData.getValue())) {
                                continue;     // "Repeated limited."
                            }

                            checkRegistryData(regData);
                        }
                    } catch (Exception e) {
                        if (!executorStoped) {
                            log.error(e.getMessage(), e);
                        }
                    }
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (Exception e) {
                        if (!executorStoped) {
                            log.error(e.getMessage(), e);
                        }
                    }
                }
            }
        });

        /**
         * 删除注册数据
         */
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                while (!executorStoped) {
                    try {
                        RegData regData = removeQueue.take();
                        if (regData != null) {
                            //删除操作
                            regDataDao.deleteDataValue(regData.getKey(), regData.getValue());
                            //检查文件状态
                            Registry fileRegistry = getFileRegistryData(regData);
                            if (fileRegistry != null && !fileRegistry.getDataList().contains(regData.getValue())) {
                                continue;   // "Repeated limited."
                            }
                            //检查注册信息并发送消息
                            checkRegistryData(regData);
                        }
                    } catch (Exception e) {
                        if (!executorStoped) {
                            log.error(e.getMessage(), e);
                        }
                    }
                }
            }
        });

        /**
         *  清理旧的注册数据 registry-data 10秒1次
         *  同步注册文件
         */
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                while (!executorStoped) {
                    try {

                        if (staticRegData != null) {
                            registryQueue.add(staticRegData);
                        }

                        regDataDao.cleanData(beatTime);

                        regDao.cleanDead();

                        int offset = 0;
                        int pagesize = 100;
                        List<String> registryDataFileList = new ArrayList<>();

                        List<Registry> registryList = regDao.pageList(offset, pagesize);
                        while (registryList != null && registryList.size() > 0) {
                            for (Registry registryItem : registryList) {
                                // default, sync from db
                                List<RegData> regDataList = regDataDao.findData(registryItem.getKey());
                                List<String> valueList = new ArrayList<String>();
                                if (regDataList != null && regDataList.size() > 0) {
                                    for (RegData dataItem : regDataList) {
                                        valueList.add(dataItem.getValue());
                                    }
                                }
                                String dataJson = FastJsonUtil.objectToJSON(valueList);

                                // 检查更新, 同步数据库
                                if (!registryItem.getData().equals(dataJson)) {
                                    registryItem.setData(dataJson);
                                    regDao.update(registryItem);
                                }

                                // 同步到本地磁盘
                                String registryDataFile = setFileRegistryData(registryItem);

                                // 注册文件加入集合
                                registryDataFileList.add(registryDataFile);
                            }

                            offset += 100;
                            registryList = regDao.pageList(offset, pagesize);
                        }

                        // 清除旧的注册数据文件
                        cleanFileRegistryData(registryDataFileList);

                    } catch (Exception e) {
                        if (!executorStoped) {
                            log.error(e.getMessage(), e);
                        }
                    }
                    try {
                        TimeUnit.SECONDS.sleep(beatTime);
                    } catch (Exception e) {
                        if (!executorStoped) {
                            log.error(e.getMessage(), e);
                        }
                    }
                }
            }
        });


    }

    @Override
    public void destroy() throws Exception {
        executorStoped = true;
        executorService.shutdownNow();
    }

    /**
     * 获取文件注册数据
     *
     * @param regData
     * @return
     */
    public Registry getFileRegistryData(RegData regData) {

        // 配置文件
        String fileName = parseRegistryDataFileName(regData.getKey());

        // 加载配置文件
        Properties prop = PropUtil.loadProp(fileName);
        if (prop != null) {
            Registry fileRegistry = new Registry();
            fileRegistry.setKey(regData.getKey()); //文件名
            fileRegistry.setData(prop.getProperty("data")); //data属性
            //data属性值解析到列表中
            fileRegistry.setDataList(JacksonUtil.readValue(fileRegistry.getData(), List.class));
            return fileRegistry;
        }
        return null;
    }

    private String parseRegistryDataFileName(String key) {
        // 注册文件名称
        String fileName = rootPath+regDataFilePath.concat(File.separator).concat(key).concat(".properties");
        return fileName;
    }

    // set
    public String setFileRegistryData(Registry registry) {
        // 文件名
        String fileName = parseRegistryDataFileName(registry.getKey());

        // 如果已注册, 直接返回路径
        Properties existProp = PropUtil.loadProp(fileName);
        if (existProp != null && existProp.getProperty("data").equals(registry.getData())) {
            return new File(fileName).getPath();
        }

        // 写入文件
        Properties prop = new Properties();
        prop.setProperty("data", registry.getData());

        PropUtil.writeProp(prop, fileName);

        log.info("setFileRegistryData: key={}, data={}", registry.getKey(), registry.getData());


        // brocast monitor client
        List<DeferredResult> deferredResultList = registryDeferredResultMap.get(fileName);
        if (deferredResultList != null) {
            registryDeferredResultMap.remove(fileName);
            for (DeferredResult deferredResult : deferredResultList) {
                deferredResult.setResult(new JsonResult<>(JsonResult.SUCCESS_CODE, "Monitor key update."));
            }
        }

        return new File(fileName).getPath();
    }

    // 清除注册客户端文件
    public void cleanFileRegistryData(List<String> registryDataFileList) {
        filterChildPath(new File(rootPath+regDataFilePath), registryDataFileList);
    }

    public void filterChildPath(File parentPath, final List<String> registryDataFileList) {
        if (!parentPath.exists() || parentPath.list() == null || parentPath.list().length == 0) {
            return;
        }
        File[] childFileList = parentPath.listFiles();
        for (File childFile : childFileList) {
            if (childFile.isFile() && !registryDataFileList.contains(childFile.getPath())) {
                childFile.delete();
                log.info("清空文件注册数据, RegData Path={}", childFile.getPath());
            }
            if (childFile.isDirectory()) {
                if (parentPath.listFiles() != null && parentPath.listFiles().length > 0) {
                    filterChildPath(childFile, registryDataFileList);
                } else {
                    childFile.delete();
                }

            }
        }

    }


}
